package com.tensai.study.java.newconcept.typereactiveprogram.reactor.sinks;

import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import reactor.core.publisher.Sinks;
import reactor.util.concurrent.Queues;

/**
 * @author Tensai
 */
@Slf4j
public class SinksManyTest {

	@Test
	public void test() throws InterruptedException {
		Sinks.Many<String> stringSinksMany = Sinks.many().multicast()
				.onBackpressureBuffer(Queues.SMALL_BUFFER_SIZE, false);

		stringSinksMany.asFlux().collectList().subscribe(System.out::println);
		for (; ; ) {
			stringSinksMany.tryEmitNext(System.currentTimeMillis()+"");
//			stringSinksMany.asFlux().subscribe();
//			log.info("关注人数： {}", stringSinksMany.currentSubscriberCount());
//			Thread.sleep(1000);
		}
	}
}
